package com.qxp.simplerpc;

import com.qxp.simplerpc.connect.RpcClientHandler;
import com.qxp.simplerpc.entity.CommonProperties;
import com.qxp.simplerpc.entity.Registry;
import com.qxp.simplerpc.exception.RpcException;
import com.qxp.simplerpc.parse.URLParse;
import com.qxp.simplerpc.utils.SimpleRpcProperties;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;

@Component
public class ZookeeperRegisterServiceImpl extends AbstractRegistryService {
    // logger
    private static final Logger logger = LoggerFactory.getLogger(ZookeeperRegisterServiceImpl.class);

    @Autowired
    public void setSimpleRpcProperties(SimpleRpcProperties simpleRpcProperties) {
        this.simpleRpcProperties = simpleRpcProperties;
    }

    // zookeeper
    private ZooKeeper zooKeeper;

    @PostConstruct
    public void initRegister() {
        initRegisterProperties();
    }

    @Override
    public void register(String url) {
        try {
            zooKeeper = new ZooKeeper(address, timeout, (WatchEvent) -> {
            });
            doRegister(url);
        } catch (Exception e) {
            logger.error("服务注册异常..", e);
        }
    }

    @Override
    protected void doRegister(String url) {
        try {
            Stat stat = zooKeeper.exists(clustername, false);
            if (stat == null) {
                zooKeeper.create(clustername, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            //创建短暂的可排序的子节点
            zooKeeper.create(stringFormatWithSplit(clustername, nodeName), url.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        } catch (Exception e) {
            logger.error("服务注册异常..", e);
        }
    }

    @Override
    public void unregister(String url) {
    }

    @Override
    public void subscribe(RpcClientHandler rpcClientHandler) {
        try {
            zooKeeper = new ZooKeeper(address, timeout, (WatchEvent) -> {
                //监听该节点的变化，如果节点出现变化，则重新获取节点下的ip和端口
                if (WatchEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged &&
                        WatchEvent.getPath().equals(clustername)) {
                    doSubscribe(rpcClientHandler);
                }
            });
            doSubscribe(rpcClientHandler);
        } catch (Exception e) {
            logger.error("zookeeper提交子节点信息失败：\n", e);
        }
    }

    @Override
    public String unsubcribe(RpcClientHandler rpcClientHandler) {
        return null;
    }

    @Override
    public List<String> listUrl() {
        List<String> list = new CopyOnWriteArrayList<>();
        try {
            zooKeeper.getChildren(clustername, true).parallelStream().forEach(childName -> {
                try {
                    byte[] dataObj = zooKeeper.getData(stringFormatWithSplit(clustername, childName), false, null);
                    list.add(new String(dataObj, StandardCharsets.UTF_8));
                } catch (Exception e) {
                    logger.error("获取zookeeper节点数据失败：\n", e);
                }
            });
        } catch (Exception e) {
            logger.error("获取zookeeper子节点失败：\n", e);
        }
        return list;
    }


}
